﻿using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Data.SqlClient;
using System.Diagnostics;
using System.IO;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using CoreLab.Oracle;
using Common.Utilities;
using Common.Utilities.Dispatcher;

namespace Workflows.Components.ETL
{
    /// <summary>
    /// 
    /// </summary>
    public class GrabDataWorker:IWorker
    {
        private Thread _workerThread;
        private GrabDataInstruction _instruction;
        private GrabDataOutput _output;
        private Guid _JobTicketID;
        private RunContext _context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }
        /// <summary></summary>
        public Thread WorkerThread
        {
            get { return _workerThread; }
        }
        /// <summary></summary>
        public InstructionBase Instruction
        {
            get { return _instruction; }
        }
        /// <summary></summary>
        public ResultBase Output
        {
            get { return _output; }
        }
        /// <summary></summary>
        public Type ComponentType
        {
            get { return typeof(GrabDataComponent); }
        }
        /// <summary></summary>
        public RunContext Context
        {
            get { return _context; }
            set { _context = value; }
        }
        /// <summary></summary>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._instruction =(GrabDataInstruction) request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._context = context;
            this._JobTicketID = ticketGuid;
            this._instruction = (GrabDataInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void DoWork()
        {
            if(this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._instruction);
            }
            this._workerThread = Thread.CurrentThread;
            this.GrabData();
        }
        /// <summary></summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._instruction);
            }
            this._workerThread = new Thread(new ThreadStart(this.GrabData));
            this._workerThread.Start();
        }
        /// <summary></summary>
        public void CancelWork()
        {
            if(this._workerThread !=null && this._workerThread.IsAlive)
            {
                this._workerThread.Abort();
                this._workerThread.Join();

                if(this.WorkerCancelledEvent !=null)
                {
                    this.WorkerCancelledEvent(this._JobTicketID, this.GetType(), this._instruction);
                }
            }
        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        #region test
        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public TestResult Test()
        {
            try
            {
                List<string> keyValues = new List<string>();
                //SqlDbType keyFieldType = SqlDbType.Int;
                TableColumn[] keyCols =
                    DbUtil.GetTableColumns(this._instruction.KeyDataProvider,
                                           this._instruction.KeyConnStr, this._instruction.KeySql);
                Debug.Assert(keyCols.Length == 1);
                //keyFieldType = keyCols[0].DbType;
                IDbConnection keyConn = null;
                if (this._instruction.KeyDataProvider == DataProviderType.Oracle)
                {
                    keyConn =
                        new OracleConnection(
                            this._instruction.KeyConnStr);
                }
                else
                {
                    keyConn =
                        new SqlConnection(this._instruction.KeyConnStr);
                }
                try
                {
                    keyConn.Open();
                    IDbCommand keyCmd = keyConn.CreateCommand();
                    keyCmd.CommandText = this._instruction.KeySql;
                    IDataReader keyReader = keyCmd.ExecuteReader();
                    while (keyReader.Read())
                    {
                        keyValues.Add(keyReader[0].ToString());
                        if(keyValues.Count>=5)
                            break;
                    }
                    keyReader.Close();
                }
                catch (Exception ex1)
                {
                    throw new Exception("Unable to retrieve keys: " + ex1.ToString());
                }
                finally
                {
                    keyConn.Close();
                }

                IDbConnection srcConn = null;
                if (this._instruction.SrcDataProvider == DataProviderType.Oracle)
                {
                    srcConn =
                        new OracleConnection(this._instruction.SrcConnStr);
                }
                else
                {
                    srcConn =
                        new SqlConnection(this._instruction.SrcConnStr);
                }
                IDbConnection tgtConn = null;
                if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                {
                    tgtConn =
                        new OracleConnection(this._instruction.TgtConnStr);
                }
                else
                {
                    tgtConn =
                        new SqlConnection(this._instruction.TgtConnStr);
                }

                // update/insert stored procedure creation
                Regex srcSqlWhereRegex = new Regex("\\s+([^\\s]+)\\s*=\\s*'?\\{0\\}'?", RegexOptions.IgnoreCase);
                string srcDataSql = this._instruction.SrcDataSql;
                Match srcSqlWhereMatch = srcSqlWhereRegex.Match(srcDataSql);
                Debug.Assert(srcSqlWhereMatch.Success);
                string srcKeyFieldName = srcSqlWhereMatch.Groups[1].Value;
                srcDataSql = srcSqlWhereRegex.Replace(srcDataSql, "");
                if(srcDataSql.ToLower().Trim().EndsWith("where"))
                {
                    srcDataSql = srcDataSql.Trim().Substring(0, srcDataSql.Trim().Length - 5);
                }

                Dictionary<string, IDataParameter> parameters = new Dictionary<string, IDataParameter>();
                TableColumn[] srcCols =
                    DbUtil.GetTableColumns(this._instruction.SrcDataProvider,
                                           this._instruction.SrcConnStr, srcDataSql);
                List<TableColumn> tgtCols = new List<TableColumn>();
                TableColumn tgtKeyCol = new TableColumn();
                foreach (TableColumn srcCol in srcCols)
                {
                    if (this._instruction.FieldMappings != null && this._instruction.FieldMappings.Count > 0)
                    {
                        foreach (GrabDataFieldMapping fieldMapping in this._instruction.FieldMappings)
                        {
                            if (fieldMapping.SrcField.ToLower() == srcCol.ColumnName.ToLower())
                            {
                                tgtKeyCol = srcCol;
                                tgtKeyCol.ColumnName = fieldMapping.TgtField;
                                break;
                            }
                        }
                    }
                    if (!string.IsNullOrEmpty(tgtKeyCol.ColumnName))
                    {
                        break;
                    }
                }
                Debug.Assert((!string.IsNullOrEmpty(tgtKeyCol.ColumnName)),
                             "source sql statement must include key field");

                foreach (TableColumn srcCol in srcCols)
                {
                    if (this._instruction.FieldMappings != null && this._instruction.FieldMappings.Count > 0)
                    {
                        foreach (GrabDataFieldMapping fieldMapping in this._instruction.FieldMappings)
                        {
                            if (fieldMapping.SrcField.ToLower() == srcCol.ColumnName.ToLower())
                            {
                                bool addToParameterCollection = false;
                                if (fieldMapping.TgtField.ToLower() == tgtKeyCol.ColumnName.ToLower())
                                {
                                    addToParameterCollection = true;
                                }
                                if (fieldMapping.Action != FieldDataPopulationAction.Ignore)
                                {
                                    addToParameterCollection = true;
                                }
                                if (addToParameterCollection)
                                {
                                    IDataParameter tgtParameter = null;
                                    if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                                    {
                                        tgtParameter =
                                            new OracleParameter(fieldMapping.TgtField,
                                                                FieldDataType.ToOracleFieldDataType(srcCol.DbType));
                                        parameters.Add(srcCol.ColumnName, tgtParameter);
                                    }
                                    else
                                    {
                                        tgtParameter = new SqlParameter(fieldMapping.TgtField, srcCol.DbType);
                                        parameters.Add(srcCol.ColumnName, tgtParameter);
                                    }
                                }

                                TableColumn tgtCol = srcCol;
                                tgtCol.ColumnName = fieldMapping.TgtField;
                                tgtCols.Add(tgtCol);
                                break;
                            }
                        }
                    }
                    else
                    {
                        TableColumn tgtCol = srcCol;
                        tgtCols.Add(tgtCol);
                        if (srcCol.ColumnName.ToLower() == srcKeyFieldName.ToLower())
                        {
                            tgtKeyCol = tgtCol;
                        }
                        IDataParameter tgtParameter = null;
                        if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                        {
                            tgtParameter =
                                new OracleParameter(srcCol.ColumnName,
                                                    FieldDataType.ToOracleFieldDataType(srcCol.DbType));
                            parameters.Add(srcCol.ColumnName, tgtParameter);
                        }
                        else
                        {
                            tgtParameter = new SqlParameter(srcCol.ColumnName, srcCol.DbType);
                            parameters.Add(srcCol.ColumnName, tgtParameter);
                        }
                    }
                }
                

                string sp_InsertName = "BulkInsert_" + this._instruction.TgtTableName;
                string sp_Insert =
                    StoredProcUtil.CreateStoredProcedureStatementForInsert(this._instruction.TgtDataProvider,
                                                                           sp_InsertName, string.Empty,
                                                                           this._instruction.TgtTableName,
                                                                           tgtCols.ToArray());
                string sp_UpdateName = "BulkUpdate_" + this._instruction.TgtTableName;

                Debug.Assert(!string.IsNullOrEmpty(tgtKeyCol.ColumnName));
                FilterField filter = new FilterField();
                filter.FieldDbDataType = new FieldDataType();
                filter.FieldDbDataType.DataType = tgtKeyCol.DbType;
                filter.FieldDbDataType.Nullible = tgtKeyCol.AllowNull;
                filter.FieldDbDataType.Precision = tgtKeyCol.Precision;
                filter.FieldDbDataType.Provider = this._instruction.TgtDataProvider;
                filter.FieldDbDataType.Scale = tgtKeyCol.Scale;
                filter.FieldDbDataType.Size = tgtKeyCol.Size;
                filter.FieldName = tgtKeyCol.ColumnName;
                string sp_Update =
                    StoredProcUtil.CreateStoredProcedureStatementForUpdate(this._instruction.TgtDataProvider,
                                                                           sp_UpdateName, string.Empty,
                                                                           this._instruction.TgtTableName,
                                                                           tgtCols.ToArray(), new FilterField[] { filter });

                StringBuilder buffer=new StringBuilder();
                try
                {
                    srcConn.Open();
                    tgtConn.Open();

                    IDbCommand srcCmd = srcConn.CreateCommand();
                    IDbCommand tgtCmd = tgtConn.CreateCommand();
                    tgtCmd.CommandType = CommandType.Text;
                    string dropInsertSp =
                        StoredProcUtil.CreateDropStoredProcedureStatement(this._instruction.TgtDataProvider,
                                                                          sp_InsertName);
                    tgtCmd.CommandText = dropInsertSp;
                    tgtCmd.ExecuteNonQuery();
                    tgtCmd.CommandText = sp_Insert;
                    tgtCmd.ExecuteNonQuery();
                    string dropUpdateSp =
                        StoredProcUtil.CreateDropStoredProcedureStatement(this._instruction.TgtDataProvider,
                                                                          sp_UpdateName);
                    tgtCmd.CommandText = dropUpdateSp;
                    tgtCmd.ExecuteNonQuery();
                    tgtCmd.CommandText = sp_Update;
                    tgtCmd.ExecuteNonQuery();

                    foreach (string keyValue in keyValues)
                    {
                        try
                        {
                            string selectSql = this._instruction.SrcDataSql.Replace("{0}", keyValue);
                            srcCmd.CommandType = CommandType.Text;
                            srcCmd.CommandText = selectSql;
                            IDataReader srcReader = srcCmd.ExecuteReader();

                            if (srcReader.Read())
                            {
                                tgtCmd.Parameters.Clear();
                                string sqlPlan = string.Empty;

                                bool exist = false;
                                tgtCmd.CommandType = CommandType.Text;
                                tgtCmd.CommandText = "select count (" + tgtKeyCol.ColumnName + ") from " +
                                                     this._instruction.TgtTableName;
                                try
                                {
                                    int count = int.Parse(tgtCmd.ExecuteScalar().ToString());
                                    if (count > 0)
                                        exist = true;
                                }
                                catch
                                {
                                    exist = false;
                                }
                                if (exist)
                                {
                                    tgtCmd.CommandType = CommandType.StoredProcedure;
                                    tgtCmd.CommandText = sp_UpdateName;
                                    sqlPlan = "exec " + sp_UpdateName;
                                }
                                else
                                {
                                    tgtCmd.CommandType = CommandType.StoredProcedure;
                                    tgtCmd.CommandText = sp_InsertName;
                                    sqlPlan = "exec " + sp_InsertName;
                                }

                                
                                foreach (string srcColName in parameters.Keys)
                                {
                                    if (!(srcReader[srcColName] is DBNull))
                                    {
                                        object fieldValue = srcReader[srcColName];
                                        IDataParameter parameter = parameters[srcColName];
                                        parameter.Value = fieldValue;
                                        tgtCmd.Parameters.Add(parameter);
                                        if(sqlPlan.IndexOf("@")>0)
                                        {
                                            sqlPlan += ",";
                                        }
                                        sqlPlan += " @" + parameter.ParameterName;
                                    }
                                }
                                // tgtCmd.ExecuteNonQuery();

                                buffer.Append(sqlPlan + Environment.NewLine);
                            }
                            srcReader.Close();
                        }
                        catch (Exception ex3)
                        {
                            if (this.WorkerErredEvent != null)
                            {
                                this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._instruction,
                                                      ex3.ToString(), false);
                            }
                        }
                    }
                }
                catch (Exception ex2)
                {
                    throw new Exception("Critical error occurred while grabbing data: " + ex2.ToString());
                }
                finally
                {
                    srcConn.Close();
                    tgtConn.Close();
                }


                return new TestResult(true, buffer.ToString());
            }
            catch(Exception ex)
            {
                return new TestResult(false,ex.ToString());
            }
        }
        #endregion

        #region work
        private void GrabData()
        {
            string appDataFolderPath =
                Path.Combine(Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData), "Workflow Data");

            string logFilePath = this._instruction.LogFilePath;
            if(string.IsNullOrEmpty(logFilePath))
            {
                logFilePath = Path.Combine(appDataFolderPath, "GrabData.log");
            }
            string errFilePath = this._instruction.ErrorFilePath;
            if(string.IsNullOrEmpty(errFilePath))
            {
                errFilePath = Path.Combine(appDataFolderPath, "GrabData.err");
            }
            if(File.Exists(logFilePath))
            {
                File.Delete(logFilePath);
            }
            if(File.Exists(errFilePath))
            {
                File.Delete(errFilePath);
            }
            

            try
            {
                int totalRecUpdated = 0;
                int totalRecScanned = 0;
                List<string> keyValues = new List<string>();
                //SqlDbType keyFieldType = SqlDbType.Int;
                TableColumn[] keyCols =
                    DbUtil.GetTableColumns(this._instruction.KeyDataProvider,
                                           this._instruction.KeyConnStr, this._instruction.KeySql);
                Debug.Assert(keyCols.Length == 1);
                //keyFieldType = keyCols[0].DbType;
                IDbConnection keyConn = null;
                if (this._instruction.KeyDataProvider == DataProviderType.Oracle)
                {
                    keyConn =
                        new OracleConnection(
                            this._instruction.KeyConnStr);
                }
                else
                {
                    keyConn =
                        new SqlConnection(this._instruction.KeyConnStr);
                }
                try
                {
                    keyConn.Open();
                    IDbCommand keyCmd = keyConn.CreateCommand();
                    keyCmd.CommandTimeout = 60 * 5;     // timeout set to 5 min to retrieve keys
                    keyCmd.CommandText = this._instruction.KeySql;
                    IDataReader keyReader = keyCmd.ExecuteReader();
                    while (keyReader.Read())
                    {
                        keyValues.Add(keyReader[0].ToString());
                    }
                    keyReader.Close();
                }
                catch (Exception ex1)
                {
                    throw new Exception("Unable to retrieve keys: " + ex1.ToString());
                }
                finally
                {
                    keyConn.Close();
                }

                IDbConnection srcConn = null;
                if (this._instruction.SrcDataProvider == DataProviderType.Oracle)
                {
                    srcConn =
                        new OracleConnection(this._instruction.SrcConnStr);
                }
                else
                {
                    srcConn =
                        new SqlConnection(this._instruction.SrcConnStr);
                }
                IDbConnection tgtConn = null;
                if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                {
                    tgtConn =
                        new OracleConnection(this._instruction.TgtConnStr);
                }
                else
                {
                    tgtConn =
                        new SqlConnection(this._instruction.TgtConnStr);
                }

                // update/insert stored procedure creation
                string srcKeyFieldName = string.Empty;
                Regex srcSqlWhereRegex = new Regex("where\\s+([^\\s]+)\\s*=\\s*'?\\{0\\}'?", RegexOptions.IgnoreCase);
                string srcDataSql = this._instruction.SrcDataSql;
                Match srcSqlWhereMatch = srcSqlWhereRegex.Match(srcDataSql);
                if(srcSqlWhereMatch.Success)
                {
                    srcKeyFieldName = srcSqlWhereMatch.Groups[1].Value;
                    srcDataSql = srcSqlWhereRegex.Replace(srcDataSql, "");
                }
                else
                {
                    srcSqlWhereRegex = new Regex("\\s+([^\\s]+)\\s*=\\s*'?\\{0\\}'?", RegexOptions.IgnoreCase);
                    srcSqlWhereMatch = srcSqlWhereRegex.Match(srcDataSql);
                    if (srcSqlWhereMatch.Success)
                    {
                        srcKeyFieldName = srcSqlWhereMatch.Groups[1].Value;
                        srcDataSql = srcSqlWhereRegex.Replace(srcDataSql, "");
                    }
                    else
                    {
                        srcSqlWhereRegex = new Regex("\\s+([^\\s]+)\\s+in\\s+'?\\(\\{0\\}\\)'?", RegexOptions.IgnoreCase);
                        srcSqlWhereMatch = srcSqlWhereRegex.Match(srcDataSql);
                        if (srcSqlWhereMatch.Success)
                        {
                            srcKeyFieldName = srcSqlWhereMatch.Groups[1].Value;
                            srcDataSql = srcSqlWhereRegex.Replace(srcDataSql, "");
                        }
                    }
                }
                srcDataSql = srcDataSql.Trim();
                if (srcDataSql.ToLower().EndsWith("where"))
                {
                    srcDataSql = srcDataSql.Substring(0, srcDataSql.Length - 5);
                }
                if (srcDataSql.ToLower().EndsWith("and"))
                {
                    srcDataSql = srcDataSql.Substring(0, srcDataSql.Length - 3);
                }

                Dictionary<string, IDataParameter> parameters = new Dictionary<string, IDataParameter>();
                TableColumn[] srcCols =
                    DbUtil.GetTableColumns(this._instruction.SrcDataProvider,
                                           this._instruction.SrcConnStr, srcDataSql);
                List<TableColumn> tgtCols = new List<TableColumn>();
                TableColumn[] allTgtCols =
                    DbUtil.GetTableColumns(this._instruction.TgtDataProvider, this._instruction.TgtConnStr, string.Empty,
                                           this._instruction.TgtTableName);
                string srcKeyColName = string.Empty;
                TableColumn tgtKeyCol = new TableColumn();
                foreach (TableColumn srcCol in srcCols)
                {
                    if (this._instruction.FieldMappings != null && this._instruction.FieldMappings.Count > 0)
                    {
                        foreach (GrabDataFieldMapping fieldMapping in this._instruction.FieldMappings)
                        {
                            if (fieldMapping.SrcField.ToLower() == srcKeyFieldName.ToLower())
                            {
                                tgtKeyCol = srcCol;
                                srcKeyColName = srcCol.ColumnName;
                                tgtKeyCol.ColumnName = fieldMapping.TgtField;
                                break;
                            }
                        }
                    }
                    if (!string.IsNullOrEmpty(tgtKeyCol.ColumnName))
                    {
                        break;
                    }
                }
                Debug.Assert((!string.IsNullOrEmpty(tgtKeyCol.ColumnName)),
                             "source sql statement must include key field");

                foreach (TableColumn tgtCol in allTgtCols)
                {
                    if (this._instruction.FieldMappings != null && this._instruction.FieldMappings.Count > 0)
                    {
                        foreach (GrabDataFieldMapping fieldMapping in this._instruction.FieldMappings)
                        {
                            if (fieldMapping.TgtField.ToLower() == tgtCol.ColumnName.ToLower())
                            {
                                bool addToParameterCollection = false;
                                if (fieldMapping.TgtField.ToLower() == tgtKeyCol.ColumnName.ToLower())
                                {
                                    addToParameterCollection = true;
                                }
                                if (fieldMapping.Action != FieldDataPopulationAction.Ignore)
                                {
                                    addToParameterCollection = true;
                                }
                                if (addToParameterCollection)
                                {
                                    IDataParameter tgtParameter = null;
                                    if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                                    {
                                        tgtParameter =
                                            new OracleParameter(fieldMapping.TgtField,
                                                                FieldDataType.ToOracleFieldDataType(tgtCol.DbType));
                                        parameters.Add(tgtCol.ColumnName, tgtParameter);
                                    }
                                    else
                                    {
                                        tgtParameter = new SqlParameter(fieldMapping.TgtField, tgtCol.DbType);
                                        parameters.Add(tgtCol.ColumnName, tgtParameter);
                                    }
                                }

                                tgtCols.Add(tgtCol);
                                break;
                            }
                        }
                    }
                }


                string sp_InsertName = "BulkInsert_" + this._instruction.TgtTableName;
                string sp_Insert =
                    StoredProcUtil.CreateStoredProcedureStatementForInsert(this._instruction.TgtDataProvider,
                                                                           sp_InsertName, string.Empty,
                                                                           this._instruction.TgtTableName,
                                                                           tgtCols.ToArray());
                string sp_UpdateName = "BulkUpdate_" + this._instruction.TgtTableName;

                Debug.Assert(!string.IsNullOrEmpty(tgtKeyCol.ColumnName));
                FilterField filter = new FilterField();
                filter.FieldDbDataType = new FieldDataType();
                filter.FieldDbDataType.DataType = tgtKeyCol.DbType;
                filter.FieldDbDataType.Nullible = tgtKeyCol.AllowNull;
                filter.FieldDbDataType.Precision = tgtKeyCol.Precision;
                filter.FieldDbDataType.Provider = this._instruction.TgtDataProvider;
                filter.FieldDbDataType.Scale = tgtKeyCol.Scale;
                filter.FieldDbDataType.Size = tgtKeyCol.Size;
                filter.FieldName = tgtKeyCol.ColumnName;
                string sp_Update =
                    StoredProcUtil.CreateStoredProcedureStatementForUpdate(this._instruction.TgtDataProvider,
                                                                           sp_UpdateName, string.Empty,
                                                                           this._instruction.TgtTableName,
                                                                           tgtCols.ToArray(), new FilterField[] { filter });
                StreamWriter logWriter = new StreamWriter(logFilePath);
                StreamWriter errWriter = new StreamWriter(errFilePath);

                bool createLocalCopyOfSourceData = false;
                string tmpTableName = "tmp_BCP_" + DateTime.Now.Ticks.ToString();

                if (this._instruction.UseBcp)
                {
                    
                    // bcp 
                    try
                    {
                        TableColumn[] bcpCols = DbUtil.GetTableColumns(this._instruction.SrcDataProvider,
                                                                       this._instruction.SrcConnStr, srcDataSql);
                        TableHelper.CreateTable(this._instruction.TgtConnStr, tmpTableName, bcpCols, false);

                        Regex dataSrcRegex = new Regex(
                            "server\\s*=\\s*([^'\"\\s;]+)||data source\\s*=\\s*([^'\"\\s;]+)",
                            RegexOptions.IgnoreCase);
                        string dbSvr1 = string.Empty;
                        string dbSvr2 = string.Empty;
                        Match dataSrcMatch = dataSrcRegex.Match(this._instruction.SrcConnStr);
                        if (dataSrcMatch.Success)
                        {
                            dbSvr1 = dataSrcMatch.Groups[1].Value;
                        }
                        dataSrcMatch = dataSrcRegex.Match(this._instruction.TgtConnStr);
                        if (dataSrcMatch.Success)
                        {
                            dbSvr2 = dataSrcMatch.Groups[1].Value;
                        }
                        if (dbSvr1.ToLower() != dbSvr2.ToLower())
                        {
                            createLocalCopyOfSourceData =
                                this.CreateLocalCopyOfDataUsingBcp(this._instruction.SrcConnStr,
                                                                   srcDataSql,
                                                                   this._instruction.TgtConnStr,
                                                                   tmpTableName,
                                                                   new string[] {srcKeyColName});
                        }
                    }
                    catch (Exception ex)
                    {
                        createLocalCopyOfSourceData = false;
                    }
                }

                try
                {
                    if(createLocalCopyOfSourceData)
                    {
                        Dictionary<string,FieldDataType> fieldsToGrab=new Dictionary<string, FieldDataType>();
                        Dictionary<string,string> src_tgt_FieldMappings=new Dictionary<string, string>();
                        Dictionary<string, TableColumn> srcColByNames=new Dictionary<string, TableColumn>();
                        foreach(TableColumn col in srcCols)
                        {
                            srcColByNames.Add(col.ColumnName,col);
                        }
                        foreach(GrabDataFieldMapping mapping in this._instruction.FieldMappings)
                        {
                            if(mapping.SrcField!=srcKeyColName)
                            {
                                TableColumn srcCol = srcColByNames[mapping.SrcField];
                                FieldDataType srcFieldDataType=new FieldDataType();
                                srcFieldDataType.DataType = srcCol.DbType;
                                srcFieldDataType.Size = srcCol.Size;
                                fieldsToGrab.Add(srcCol.ColumnName,srcFieldDataType);
                                src_tgt_FieldMappings.Add(mapping.SrcField,mapping.TgtField);
                            }
                        }

                        srcConn.ConnectionString = this._instruction.TgtConnStr;
                        this.UpdateFields(this._instruction.TgtConnStr,tmpTableName,
                            srcKeyColName,tgtKeyCol.DbType,tgtKeyCol.ColumnName,
                            this._instruction.TgtTableName,fieldsToGrab, 
                            src_tgt_FieldMappings);
                        if (createLocalCopyOfSourceData)
                        {
                            DbUtil.DropTable(this._instruction.TgtConnStr, tmpTableName);
                        }
                    }
                    else
                    {
                        srcConn.Open();
                        tgtConn.Open();

                        IDbCommand srcCmd = srcConn.CreateCommand();
                        IDbCommand tgtCmd = tgtConn.CreateCommand();

                        srcCmd.CommandTimeout = 60 * 15;
                        tgtCmd.CommandTimeout = 60 * 15;

                        tgtCmd.CommandType = CommandType.Text;
                        string dropInsertSp =
                            StoredProcUtil.CreateDropStoredProcedureStatement(this._instruction.TgtDataProvider,
                                                                              sp_InsertName);
                        tgtCmd.CommandText = dropInsertSp;
                        tgtCmd.ExecuteNonQuery();
                        tgtCmd.CommandText = sp_Insert;
                        tgtCmd.ExecuteNonQuery();
                        string dropUpdateSp =
                            StoredProcUtil.CreateDropStoredProcedureStatement(this._instruction.TgtDataProvider,
                                                                              sp_UpdateName);
                        tgtCmd.CommandText = dropUpdateSp;
                        tgtCmd.ExecuteNonQuery();
                        tgtCmd.CommandText = sp_Update;
                        tgtCmd.ExecuteNonQuery();

                        foreach (string keyValue in keyValues)
                        {
                            IDataReader srcReader = null;
                            try
                            {
                                if (srcConn.State != ConnectionState.Open)
                                {
                                    srcConn.Open();
                                }
                                totalRecScanned++;
                                string selectSql = this._instruction.SrcDataSql.Replace("{0}", keyValue);
                                srcCmd.CommandType = CommandType.Text;
                                if (createLocalCopyOfSourceData)
                                {
                                    int fromPos = selectSql.ToLower().IndexOf("from");
                                    if (fromPos > 0)
                                    {
                                        int wherePos = selectSql.ToLower().IndexOf("where", fromPos);
                                        if (wherePos > 0)
                                        {
                                            selectSql = selectSql.Substring(0, fromPos) + " from " + tmpTableName + " " +
                                                        selectSql.Substring(wherePos);
                                        }
                                    }
                                }

                                srcCmd.CommandText = selectSql;
                                srcReader = srcCmd.ExecuteReader();

                                if (srcReader.Read())
                                {
                                    tgtCmd.Parameters.Clear();

                                    bool skip = false;
                                    bool exist = false;
                                    tgtCmd.CommandType = CommandType.Text;
                                    string selectCountSql = "select count (" + tgtKeyCol.ColumnName + ") from " +
                                                         this._instruction.TgtTableName + " where " + tgtKeyCol.ColumnName +
                                                         "=" +
                                                         FieldDataType.ToMssqlDataValue(tgtKeyCol.DbType, keyValue);
                                    if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                                    {
                                        selectCountSql = "select count (" + tgtKeyCol.ColumnName + ") from " +
                                                         this._instruction.TgtTableName + " where " + tgtKeyCol.ColumnName +
                                                         "=" +
                                                         FieldDataType.ToOracleDataValue(tgtKeyCol.DbType, keyValue);
                                    }
                                    tgtCmd.CommandText = selectCountSql;
                                    try
                                    {
                                        int count = int.Parse(tgtCmd.ExecuteScalar().ToString());
                                        if (count > 0)
                                            exist = true;
                                    }
                                    catch
                                    {
                                        exist = false;
                                    }
                                    if (exist)
                                    {
                                        if (!string.IsNullOrEmpty(this._instruction.NonNullFieldSkipUpdate))
                                        {
                                            tgtCmd.CommandText = "select count (*) from " + this._instruction.TgtTableName +
                                                                 " where " + this._instruction.NonNullFieldSkipUpdate +
                                                                 " is not null and " +
                                                                 tgtKeyCol.ColumnName + "=" +
                                                                 FieldDataType.ToMssqlDataValue(tgtKeyCol.DbType, keyValue);
                                            if (this._instruction.TgtDataProvider == DataProviderType.Oracle)
                                            {
                                                tgtCmd.CommandText = "select count (*) from " + this._instruction.TgtTableName +
                                                                 " where " + this._instruction.NonNullFieldSkipUpdate +
                                                                 " is not null and " +
                                                                 tgtKeyCol.ColumnName + "=" +
                                                                 FieldDataType.ToOracleDataValue(tgtKeyCol.DbType, keyValue);
                                            }
                                            tgtCmd.CommandType = CommandType.Text;
                                            try
                                            {
                                                int skipCount = int.Parse(tgtCmd.ExecuteScalar().ToString());
                                                if (skipCount > 0)
                                                    skip = true;
                                            }
                                            catch
                                            {
                                                skip = false;
                                            }
                                        }

                                        tgtCmd.CommandType = CommandType.StoredProcedure;
                                        tgtCmd.CommandText = sp_UpdateName;
                                        logWriter.WriteLine(
                                            string.Format("{0} updating record# {1}: {2}", DateTime.Now.ToString(),
                                                          totalRecScanned, keyValue));
                                        logWriter.Flush();
                                    }
                                    else
                                    {
                                        tgtCmd.CommandType = CommandType.StoredProcedure;
                                        tgtCmd.CommandText = sp_InsertName;
                                        logWriter.WriteLine(
                                            string.Format("{0} inserting record# {1}: {2}", DateTime.Now.ToString(),
                                                          totalRecScanned, keyValue));
                                        logWriter.Flush();
                                    }
                                    if (!skip)
                                    {
                                        foreach (string tgtColName in parameters.Keys)
                                        {
                                            string srcColName = tgtColName;
                                            if (this._instruction.FieldMappings != null &&
                                                this._instruction.FieldMappings.Count > 0)
                                            {
                                                foreach (
                                                    GrabDataFieldMapping fieldMapping in this._instruction.FieldMappings)
                                                {
                                                    if (fieldMapping.TgtField.ToLower() == tgtColName.ToLower())
                                                    {
                                                        srcColName = fieldMapping.SrcField;
                                                        break;
                                                    }
                                                }
                                            }

                                            object fieldValue = srcReader[srcColName];
                                            IDataParameter parameter = parameters[tgtColName];
                                            parameter.Value = fieldValue;
                                            tgtCmd.Parameters.Add(parameter);
                                        }
                                        tgtCmd.ExecuteNonQuery();
                                        totalRecUpdated++;
                                    }
                                }
                                srcReader.Close();

                                if (totalRecScanned % 10 == 0)
                                {
                                    if (this.WorkerStatusChangedEvent != null)
                                    {
                                        string msg =
                                            string.Format(
                                                "Populating data to table {0}, scanned {1} of {2}, {3} records updated",
                                                this._instruction.TgtTableName, totalRecScanned, keyValues.Count,
                                                totalRecUpdated);
                                        int percent = (totalRecScanned * 100) / keyValues.Count;
                                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._instruction,
                                                                     this.WorkflowName, this.ComponentName, msg,
                                                                      percent);
                                    }
                                }
                            }
                            catch (Exception ex3)
                            {
                                if (srcReader != null && srcReader.IsClosed == false)
                                {
                                    srcReader.Close();
                                }

                                if (this.WorkerErredEvent != null)
                                {
                                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._instruction,
                                                          ex3.ToString(),false);
                                }

                                errWriter.WriteLine("{0} Failed to update record {1}: {2}", DateTime.Now.ToString(),
                                                    keyValue, ex3.ToString());
                                errWriter.Flush();
                            }
                        }

                        if (createLocalCopyOfSourceData)
                        {
                            DbUtil.DropTable(this._instruction.TgtConnStr, tmpTableName);
                        }
                    }
                    
                }
                catch(Exception ex2)
                {
                    throw new Exception("Critical error occurred while grabbing data: "+ex2.ToString());
                }
                finally
                {
                    srcConn.Close();
                    tgtConn.Close();
                    logWriter.Flush();
                    logWriter.Close();
                    errWriter.Flush();
                    errWriter.Close();
                }

                this._output =
                    new GrabDataOutput(this._instruction.OwnerComponentID, this._JobTicketID, totalRecUpdated,
                                       logFilePath, errFilePath);
                if(this.WorkerFinishedEvent !=null)
                {
                    this.WorkerFinishedEvent(this._JobTicketID, this.GetType(), this._instruction, this._output);
                }
            }
            catch(Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._instruction, ex.ToString(),true);
                }
            }
        }

        private void LoadDataUsingSqlCursor(string bcpTmpTableName)
        {
            
        }

        private void UpdateFields(string tgtConnStr, string bcpTmpTblName, 
            string srcKeyFieldName, SqlDbType keyDbType,
            string tgtKeyFieldName, string tgtTblName, 
            Dictionary<string, FieldDataType> fieldsToGrab,
            Dictionary<string,string> srcFields_TgtFields_Mappings)
        {
            string[] fields=new string[fieldsToGrab.Count];
            fieldsToGrab.Keys.CopyTo(fields,0);
            SqlConnection conn = new SqlConnection(tgtConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                FieldDataType keyDataType=new FieldDataType();
                keyDataType.DataType = keyDbType;
                keyDataType.Size = FieldDataType.GetFieldDefaultSize(keyDbType);
                StringBuilder buffer=new StringBuilder();
                buffer.AppendLine("update " + tgtTblName);
                buffer.AppendLine("set ");
                for(int i=0;i<fields.Length;i++)
                {
                    string fieldName = fields[i];
                    buffer.AppendLine(srcFields_TgtFields_Mappings[fieldName]+ "=b."+fieldName);
                    if (i < fields.Length - 1)
                    {
                        buffer.Append(",");
                    }
                }
                buffer.AppendLine("from " + tgtTblName + " a, " + bcpTmpTblName + " b");
                buffer.AppendLine("where a." + tgtKeyFieldName + "=b." + srcKeyFieldName + ";");
                cmd.CommandText = buffer.ToString();
                cmd.CommandTimeout = 0;
                cmd.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to update fields "+ex.ToString());
            }
            finally
            {
                conn.Close();
            }

        }

        private bool CreateLocalCopyOfDataUsingBcp(
            string remoteConnStr, string selectSql, 
            string localConnStr, string localTblName, 
            string[] columnsToBeIndexed)
        {
            bool bcpSucceed = false;
            SqlConnection conn1=new SqlConnection(remoteConnStr);
            
            SqlBulkCopy bcp = null;
            try
            {
                conn1.Open();
                SqlCommand cmd1 = conn1.CreateCommand();
                cmd1.CommandText = selectSql;
                cmd1.CommandTimeout = 0;
                SqlDataReader reader = cmd1.ExecuteReader();
                bcp=new SqlBulkCopy(localConnStr);
                bcp.DestinationTableName = localTblName;
                if(this.WorkerStatusChangedEvent !=null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._instruction, this.WorkflowName, this.ComponentName,
                                                  "Creating a local copy of source data using bcp...", 10);
                }
                bcp.BulkCopyTimeout = 0;
                bcp.WriteToServer(reader);
                reader.Close();
                bcpSucceed = true;

                if (this.WorkerStatusChangedEvent != null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._instruction, this.WorkflowName, this.ComponentName,
                                                  "Creating indexes for local copy of source data using bcp...", 15);
                }
                if(columnsToBeIndexed !=null && columnsToBeIndexed.Length>0)
                {
                    foreach(string colName in columnsToBeIndexed)
                    {
                        string idxName = "idx_" + localTblName + "_" + colName;
                        DbUtil.AddIndex(localConnStr, localTblName, colName, idxName);
                    }
                }
            }
            catch(Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
                throw new Exception("Unable to bcp: "+ex.ToString());
            }
            finally
            {
                bcp.Close();
                conn1.Close();
            }
            return bcpSucceed;
        }
        #endregion
    }
}
